
//Copyright 1997-2009 Syrinx Development, Inc.
//This file is part of the Syrinx Web Application Framework (SWAF).
// == BEGIN LICENSE ==
//
// Licensed under the terms of any of the following licenses at your
// choice:
//
//  - GNU General Public License Version 3 or later (the "GPL")
//    http://www.gnu.org/licenses/gpl.html
//
//  - GNU Lesser General Public License Version 3 or later (the "LGPL")
//    http://www.gnu.org/licenses/lgpl.html
//
//  - Mozilla Public License Version 1.1 or later (the "MPL")
//    http://www.mozilla.org/MPL/MPL-1.1.html
//
// == END LICENSE ==
using System;
using System.Diagnostics;

using Swaf;
using Swaf.BizRule;
using Swaf.Call;
using Swaf.Container;
using Swaf.Scheduler;
using System.Threading;

namespace Swaf.Messaging
{
	/// <summary>
	/// Summary description for MessagingRules.
	/// </summary>
	public class MessagingRules
	{
		private static long s_uniqueInteger=0;

		public void CsfMessagingMonitorQueue(IApplication app, IRuleContext context, QueueConnectionInfo info, ICall observer)
		{
			CsfMessagingMonitorQueue(app, context, info, observer, null);
		}
		/// <summary>
		/// This is a 
		/// </summary>
		/// <param name="app"></param>
		/// <param name="context"></param>
		/// <param name="info"></param>
		/// <param name="observer"></param>
		/// <param name="schedulingFrequency"></param>
		public void CsfMessagingMonitorQueue(IApplication app, IRuleContext context, QueueConnectionInfo info, ICall observer, string delayTime)
		{
			Debug.Assert(info != null);
			Debug.Assert(observer != null);

			string statusKey = info.QueueName.Replace('.','-');
			try
			{
				bool shouldStop = beenAskedToStop(app, statusKey);
				if(!shouldStop)
				{
					if(delayTime == null || delayTime.Length == 0)
					{
						app.appData[statusKey] = "running";
						context.put("queueInfo", info);
						context.put("observers", observer);
						context.put("msgSys", app.factories["MessagingSystems"].createFromNickName(info.MessagingSystemId));
						if (!info.ReadToEnd)
						{
							long maxMessagesToRead = info.createConnection().MessageCount;
							context.put("MaxMessagesToRead", maxMessagesToRead);
							context.put("NumReadsInitiated", (long) 1);
						}
						IInitialRequest mainQueueRequest = context.RuleGroup.MainRequest;
						mainQueueRequest.MinWorkers = info.MinThreads;
						mainQueueRequest.MaxWorkers = info.MaxThreads;
						IRuleGroup group = app.ruleMgr.executeAsync(context, "CsfMessagingProcessSingleMessage", null, "", new object[]{info, observer});
						group.waitForCompletion(true);
					}
					else
					{
						scheduleMonitoring(app, delayTime, info, observer);
					}
				}
				else
				{
					app.Log.println("Queue {0} has been requested to shutdown.", info.QueueName);
				}
			}
			catch(Exception e)
			{
				app.Log.println("Exception while trying to process message on Queue {0}: {1}", info.QueueName, e.Message);
			}

			try
			{
				bool shouldStop = beenAskedToStop(app, statusKey);

				if(!shouldStop && info.ScheduleInterval != null && info.ScheduleInterval.Length != 0)
				{
					scheduleMonitoring(app, info.ScheduleInterval, info, observer);
				}
				else
				{
					app.Log.println("Queue {0} has been shutdown.", info.QueueName);
					app.appData[statusKey] = "stopped";
					app.Log.println("Not rescheduling Monitor Queue {0}", info.QueueName);
				}
				
			}
			catch(Exception e)
			{
				app.Log.println("Exception while trying to reschedule Monitor Queue {0}: {1}", info.QueueName, e.Message);
			}
		}

		private void scheduleMonitoring(IApplication app, string baseTime, QueueConnectionInfo info, ICall observer)
		{
			//create a new task in the scheduler for this MQ listener.
			IScheduleManager scheduleManager= (IScheduleManager)app.appData["sysObjs-Scheduler"];
			Debug.Assert(scheduleManager != null);

			//TODO: Beef up validation and error handling of info.ScheduleInterval because it can easily have bad data in it.
			string freq = "in " + baseTime;
			ICall ruleCall = new RuleMgrCall(info.UniqueId , "CsfMessagingMonitorQueue", info, observer, string.Empty);
			scheduleManager.createScheduledTask(info.QueueName, freq, ruleCall);
			string statusKey = info.QueueName.Replace('.','-');
			app.appData[statusKey] = "scheduled";
		}

		private bool beenAskedToStop(IApplication app, string id)
		{
			try
			{
				string info = (string)app.appData[id];
				return (info != null && 
					(String.Compare(info, "stop", true) == 0 || String.Compare(info, "stopped", true) == 0));
			}
			catch
			{
				return false;
			}
		}

		public bool CsfMessagingProcessSingleMessage(IApplication app, IRuleContext context, QueueConnectionInfo info, ICall observer)
		{
			bool gotMessage = false;

			if (! info.ReadToEnd)
			{
				// The processing for this queue is set to only
			}
			//Get Message
			IMessagingSystem msgSys = MessagingSystemBase.getThreadSafeMessagingSystem(info.MessagingSystemId);
			IMessage msg
				= null;
			try
			{
				msgSys.beginTrans(info);
				msg = msgSys.receive(info, info.MaxReadMilliseconds);
			}
			catch(Exception e)
			{
				app.Log.println("Exception while trying to receive message from queue {0}: {1}", info.QueueName, e.ToString());
				msg = null;
			}

			if(msg != null)
			{
				gotMessage = true;
				try
				{
					observer.call(msg);
					msgSys.commit(info);
				}
				catch(Exception e)
				{
					app.Log.println(e.ToString());
					msgSys.rollback(info);
				}
				finally
				{
					if(msg is IDisposable)
					{
						((IDisposable)msg).Dispose();
					}
				}

				if(!beenAskedToStop(app, info.QueueName.Replace('.','-')))
				{
					// In order to prevent multiple threads from calculating the # of additional rule calls to make, 
					// and to prevent recalculation before rule from the parent context spawns all messages
					// we lock on the
					//single object being used by the bizrule package in order to lock with.

					lock (context.RuleGroup.MainRequest)
					{
						uint numTimes = calcNumRulesNeeded(app, context);
						
						long maxMessagesToRead = 0;
						long numReadsInitiated = 0;
						if (! info.ReadToEnd)
						{
							// If we are only reading through the queue once, we need to keep track of how
							// many messages we've read so far
							numReadsInitiated = (long) context.get("NumReadsInitiated");

							// Retrieve the Max number of messages that should be received for this request
							maxMessagesToRead = (long) context.get("MaxMessagesToRead");
						}
				
						for(uint p = 0; p < numTimes; ++p)
						{
							if (! info.ReadToEnd)
							{
								// Queue reading should only be done once through the queue, if any new messages have
								// arrived, they will wait until after queue reading has been rescheduled
								if (numReadsInitiated >= maxMessagesToRead)
								{
									app.Log.println("Processed through once, do not request more rule executions for Queue: {0}", info.QueueName);

									// Ensure we don't launch anymore reads, because we are 
									break;
								}
								else
								{
									// Allowing another read to be initiated
									++numReadsInitiated;
									context.put("NumReadsInitiated", numReadsInitiated);
								}
							}

							// Initiate a Read from the Queue
							context.RuleGroup.execute(context, "CsfMessagingProcessSingleMessage", null, "", new object[]{info, observer});
						}
					}
				}
				else
				{
					// In the else because beenAskedToStop is true
					app.Log.println("Queue {0} has been requested to shutdown, no more readers will be launched.", info.QueueName);
				}
			}
			else
			{
				//We have a null message but still have an open transaction.  This
				// will block any other calls to this q manager.
				msgSys.rollback(info);
			}

			return gotMessage;
		}

		public void CsfMessagingStopMonitoringQueue(IApplication app, IRuleContext context, IMessagingConnectionInfo queueInfo)
		{
			app.appData[queueInfo.QueueName.Replace('.','-')] = "stop";
		}

		private uint calcNumRulesNeeded(IApplication app, IRuleContext context)
		{
			uint numTimes = 1;
			IInitialRequest irq = context.RuleGroup.MainRequest;
			
			//This code will execute
			int maxWorkers = irq.MaxWorkers;
			if(maxWorkers == -1)
			{
				maxWorkers = 20;
			} //TODO: This should be set to some other value;

			// Total to add needs to look at Enqueued count AND Active Rule Count
			uint curRules = irq.EnqueuedRuleCount + irq.ActiveRuleCount;
			if (curRules < maxWorkers)
			{
				// Since we are now taking into consideration, the active rules, this thread needs to
				// add a replacement for itself...that's why the +1
				numTimes = (uint)maxWorkers - curRules + 1;
			}
			else
			{
				numTimes = 1;
			}

			return numTimes;
		}
	}
}
